package com.test.demo.service;

import com.test.demo.mapper.UserMapper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class UserUpdateService {
    @Autowired
    private UserMapper userMapper;

    @Autowired
    private UserService userService;

    public Boolean update(Long id) throws Exception {
        userService.sendUserInfo(id);
        //创建mq消费对象
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("rocket-demo");
        //设置nameSrv地址：端口
        defaultMQPushConsumer.setNamesrvAddr("192.168.254.129:9876");
        defaultMQPushConsumer.setInstanceName("consumer");
        //设置topic 和 tag
        defaultMQPushConsumer.subscribe("testTopic","testTag");

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    userMapper.updateById(id);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        defaultMQPushConsumer.start();
        System.out.println("consumer is start");
        return true;
    }
}
